package com.bigdata.spark.util

import java.util

import org.apache.kafka.common.TopicPartition
import org.apache.spark.streaming.kafka010.OffsetRange
import redis.clients.jedis.Jedis

/**
  * Author: Felix
  * Date: 2020/10/23
  * Desc: 维护偏移量的工具类
  */
object OffsetManagerUtil {

  //从Redis中获取偏移量，Rdis中保存的是Map，这个Map的key是："offset:" + topic + ":" + groupId。key：partitionId value：untilOffset
  // type:hash   key: offset:topic:groupId   field:partition   value: 偏移量
  //返回值是查看kafka工具类里，带偏移量的创建连接方法知道的。
  def getOffset(topic:String,groupId:String): Map[TopicPartition,Long]={
    //获取客户端连接
    val jedis: Jedis = MyRedisUtil.getJedisClient()
    //拼接操作redis的key，value是Map。     offset:topic:groupId
    var offsetKey = "offset:" + topic + ":" + groupId


    //redis> HSET myhash field1 "Hello"
    //(integer) 1
    //redis> HSET myhash field2 "World"
    //(integer) 1
    //redis> HGETALL myhash
    //1) "field1"
    //2) "Hello"
    //3) "field2"
    //4) "World"
    //redis>

    //获取当前消费者组消费的主题  对应的分区以及偏移量
    val offsetMap: util.Map[String, String] = jedis.hgetAll(offsetKey)
    //关闭客户端
    jedis.close()
    //offsetKey是表名
    //map里是这个表里的所有 字段及字段值 分区及偏移量！是个


    //将java的map转换为scala的map
    import scala.collection.JavaConverters._
    val oMap: Map[TopicPartition, Long] = offsetMap.asScala.map {
      case (partition, offset) => {
        println("读取分区偏移量：" + partition + ":" + offset)
        //返回值//Map[TopicPartition,Long]
        (new TopicPartition(topic, partition.toInt), offset.toLong)
      }
    }.toMap
    oMap
  }


  //offsetRanges是数组的原因：一个主题多个分区的偏移量都要保存。以partitionId区分。
  //offsetRanges是主方法里获取RDD偏移量时获取的东西。


  //将偏移量信息保存到Redis中.会对每个分区的untilOffset偏移量进行保存。
  def saveOffset(topic: String, groupId: String, offsetRanges: Array[OffsetRange]): Unit = {
    //拼接redis中操作偏移量的key
    var offsetKey = "offset:" + topic + ":" + groupId
    //定义java的map集合，用于存放每个分区对应的偏移量
    val offsetMap: util.HashMap[String, String] = new util.HashMap[String,String]()
    //对offsetRanges进行遍历，将数据封装offsetMap
    for (offsetRange <- offsetRanges) {                        //Kafka有多少个分区循环多少次。放入Redis之后就有多少个键值。
      val partitionId: Int = offsetRange.partition
      val fromOffset: Long = offsetRange.fromOffset
      val untilOffset: Long = offsetRange.untilOffset

      offsetMap.put(partitionId.toString,untilOffset.toString)     //redis中存储的是partitionId+untilOffset

      println("保存分区" + partitionId + ":" + fromOffset + "----->" + untilOffset)
    }
    val jedis: Jedis = MyRedisUtil.getJedisClient()
    jedis.hmset(offsetKey,offsetMap)//redis存Map
    jedis.close()
  }

}
